package com.imooc.producer.b_order;

import com.utils.Const;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2024/4/10-2024
 * @Description: 这里用一句话描述这个类的作用
 */
public class Consumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
        consumer.setNamesrvAddr(Const.NAME_SERVER_SINGLE_LOCAL);

        consumer.subscribe("all_orderly_topic","*");
        //consumer.subscribe("partition_orderly_topic","*");

        consumer.setMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                context.setAutoCommit(true);
                  try {
                      for (MessageExt messageExt : msgs) {
                          System.out.println(new String(messageExt.getBody()));
                      }
                  }catch (Exception e){
                      e.printStackTrace();
                      //意思是等一会，一会再处理这批消息，而不是直接放到重试队列中
                      return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;

                      //在并发消费的时候，还有一个 RECONSUME_LATER。此时是重回队列，然后再从队列中取出，再进行消费。
                      //return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                  }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });


    }
}